package org.jgroups.tests;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.stack.Protocol;
import org.jgroups.tests.TestJSender.MessagePayload;
import org.jgroups.protocols.FRAG3;
import org.jgroups.protocols.FRAG2;
import org.jgroups.protocols.pbcast.NAKACK;
import org.jgroups.util.Util;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Vector;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: May 11, 2009
 * Time: 10:15:20 AM
 * To change this template use File | Settings | File Templates.
 */
public class TestJReceiver extends ReceiverAdapter {
    private static Log logger = LogFactory.getLog(TestJReceiver.class);
   private FileOutputStream sentLog;
    private FileOutputStream recvLog;
    
    int count ;
    AtomicInteger no ;
    FRAG2 frag2;
    NAKACK nakack;

    public TestJReceiver(int count) {
        this.count = count;
        this.no = new AtomicInteger(0);
        try {
        	sentLog = new FileOutputStream(new File("sent.log"),true);
        	recvLog = new FileOutputStream(new File("recv.log"),true);
        } catch(FileNotFoundException fnfe) {
        	logger.error("msgs.log cannot be opened");
        }        
    }

    public void receive(Message msg) {
        if ( msg == null ) {
            logger.warn("Mesasge is null ");
        }
        else {
            logger.info("Receive message source=" +msg.getSrc().toString()+" length="+ msg.getLength()
                    +" destination="+ msg.getDest());
        }
        int i = no.getAndIncrement();
        if ( i % count == (count -1 ) ) {
            logger.info("FRAG2 " + frag2.toString() );
            logger.info("NAKACK " + nakack.toString() );
        }
           try {
            	logger.info("about to read payload");
            MessagePayload mpl = (MessagePayload) Util.objectFromByteBuffer(msg.getBuffer());
            synchronized(recvLog) {
            	recvLog.write((Long.toString(mpl.id) + "\n").getBytes());
            	recvLog.flush();
            }
            } catch(Exception e) {
            	logger.error("PRoblem writing id");
            }        

    }

    public static void help() {
          System.out.println("java stress.org.jgroups.test.TestSender -udp ./udp.xml -cluster test-1 -count 100 size -size 10");
    }


    public static NAKACK getNAKACK(JChannel channel) {
        Vector<Protocol> vectors = channel.getProtocolStack().getProtocols();
        for (Protocol p : vectors ) {
            if ( p.getName().equals("NAKACK")) return  (NAKACK) p;
        }
        return null;
    }

    public static void main(String[] args) throws Exception {
        String[] defaults = { "./udp.xml","test-1" , "100" , "20", "2000" };

        for ( int i=0; i < args.length ; ) {
            if ( args[i].equals("-udp")) defaults[0] = args[++i] ;
            else if ( args[i].equals("-cluster")) defaults[1] = args[++i] ;
            else if ( args[i].equals("-count")) defaults[2] = args[++i] ;
            else if ( args[i].equals("-size")) defaults[3] = args[++i] ;
            else if ( args[i].equals("-times")) defaults[4] = args[++i] ;
            else if ( args[i].equals("-help")) help() ;
            else {
               help();
            }
            i ++;
        }
        int count = Integer.valueOf( defaults[2]);
        System.setProperty("java.net.preferIPv4Stack", "true");
        logger.info("java.net.preferIPv4Stack ="+ System.getProperty("java.net.preferIPv4Stack"));
        JChannel channel = new JChannel(new File(defaults[0]));
        TestJReceiver tr = new TestJReceiver(count);
        channel.setReceiver(tr);
        FRAG2 frag2 = (FRAG2) channel.getProtocolStack().getTopProtocol();
        tr.frag2 = frag2;
        tr.nakack = getNAKACK( channel) ;
        channel.connect(defaults[1]);
        logger.info(tr.nakack.toString());
    }

}
